package com.at.process13;

import com.at.bean.WaterSensor;
import com.at.functions5.WaterSensorMapFunction3;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/29 15:46
 */
public class KeyedProcessTimerDemo1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 7777)
                .map(new WaterSensorMapFunction3());

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L);

        SingleOutputStreamOperator<WaterSensor> sensorDSWithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);

        KeyedStream<WaterSensor, String> sensorKS = sensorDSWithWatermark.keyBy(WaterSensor::getId);

        // TODO Process:keyed
        sensorKS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    /**
                     * 来一条数据，调用一次
                     * @param value The input value.
                     * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
                     *     {@link TimerService} for registering timers and querying the time. The context is only
                     *     valid during the invocation of this method, do not store it.
                     * @param out The collector for returning result values.
                     * @throws Exception
                     */
                    @Override
                    public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        // 数据中提取出来的事件时间，没有就是null
                        String currentKey = ctx.getCurrentKey();

                        // TODO 1、定时器注册
                        TimerService timerService = ctx.timerService();

                        // 1、案例： 事件事件
                        Long currentEventTIme = ctx.timestamp();
                        timerService.registerEventTimeTimer(currentEventTIme+5000L);
                        System.out.println("当前key=" + currentKey + ",时间是" + currentEventTIme + ",注册了一个5s的定时器");

                        // 2、案例： 处理时间
//                        long currentTs = timerService.currentProcessingTime();
//                        timerService.registerProcessingTimeTimer(currentTs + 5000L);
//                        System.out.println("当前key=" + currentKey + ",时间是" + currentTs + ",注册了一个5s后的定时器");

                        // 3、获取当前 Process的 Watermark
                        long currentWatermark = timerService.currentWatermark();
                        System.out.println("当前数据=" + value + ",当前Watermark=" + currentWatermark);

                        // 注册多少钱： 处时间、事件时间
//                        timerService.registerProcessingTimeTimer();
//                        timerService.registerEventTimeTimer(5000L);
                        // 删除定时器： 事件时间、处理时间
//                        timerService.deleteEventTimeTimer();
//                        timerService.deleteProcessingTimeTimer();

                        // 获取当前时间进展： 处理时间-就是系统时间；  事件时间-当前Watermark
//                        timerService.currentProcessingTime();
//                        timerService.currentWatermark()
                    }

                    /**
                     * TODO 2、时间进展到定时器注册的时间，调用该方法
                     * @param timestamp 当前时间进展
                     * @param ctx 上下文
                     * @param out 采集器
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);

                        System.out.println("当前key=" + ctx.getCurrentKey() + ",现在时间是 " + timestamp + " 定时器触发");
                    }
                }
        ).print();

        env.execute();
    }
}
/**
 * TODO 定时器
 * 1、keyed才有
 * 2、事件时间定时器，通过Watermark来触发的
 *    watermark >= 注册的时间
 *    注意： Watermark = 当前最大事件时间 - 等待时间 -1ms， 因为 -1ms，所以会推迟一条数据
 *          比如，5s的定时器，
 *          如果 等待=3s， Watermark = 8s - 3s - 1ms = 4999ms，不会触发5s的定时器
 *          需要 Watermark = 9s - 3s - 1ms = 5999ms，才能去触发5s的定时器
 * 3、在Process中获取当前warkmark，显示是上一次的Watermark
 *      =》 因为process 还没接收到这条数据 对应生成的新 Watermark
 */